package com.shujia.flink.core

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time

object Demo3EventTIme {
  def main(args: Array[String]): Unit = {


    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    /**
      * 每一个并行度中都会有一个水位线，如果时多线程，测试不好看结果
      *
      */
    env.setParallelism(1)

    //修改flink时间模式为事件时间
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)


    val ds: DataStream[String] = env.socketTextStream("master", 8888)

    /*
001,zhsngsan,1599810953000
001,zhsngsan,1599810954000
001,zhsngsan,1599810955000
001,zhsngsan,1599810957000
001,zhsngsan,1599810956000
001,zhsngsan,1599810961000
001,zhsngsan,1599810965000
001,zhsngsan,1599810966000
001,zhsngsan,1599810966000
001,zhsngsan,1599810966000
001,zhsngsan,1599810966000
001,zhsngsan,1599810966000
001,zhsngsan,1599810970000
002,lisi,1599810955000
      */

    val evends: DataStream[Event] = ds.map(line => {
      val split: Array[String] = line.split(",")
      Event(split(0), split(1), split(2).toLong, 1)
    })

      //设置数据哪一个列时事件事件,  水位线等位于最新数据的事件
      //      .assignAscendingTimestamps(_.ts)

      //指定时间列和最大允许延迟的时间
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[Event](Time.seconds(5)) {
      override def extractTimestamp(element: Event): Long = {
        element.ts
      }
    })


    evends
      .keyBy(_.id)
      .timeWindow(Time.seconds(5))
      .sum("c")
      .print()

    env.execute()


  }

  case class Event(id: String, name: String, ts: Long, c: Int)

}


